package com.leeyu.kafka.selfserial;

import com.sun.xml.internal.ws.encoding.soap.DeserializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class UserDeserializer implements Deserializer<User> {

    @Override
    public User deserialize(String topic, byte[] bytes) {
        try {
            if (bytes == null) {
                return null;
            }
            if (bytes.length < 8) {
                throw new DeserializationException("User deserialization size error");
            }

            ByteBuffer buffer = ByteBuffer.wrap(bytes);
            int id;
            String name;
            int nameSize;
            id = buffer.getInt();
            nameSize = buffer.getInt();
            byte [] nameBytes = new byte[nameSize];
            buffer.get(nameBytes);
            name = new String(nameBytes, StandardCharsets.UTF_8);
            return new User(id, name);
        } catch (Exception e) {
            throw new DeserializationException("Deserialization User error", e);
        }
    }
}
